package com.yeming.socket.producer

import java.io.{ObjectInputStream, ObjectOutputStream}
import java.net.{ServerSocket, Socket}
import java.util.{Properties, UUID}

import com.google.gson.{Gson, GsonBuilder}
import com.yeming.socket.`case`.LoggingData
import com.yeming.socket.utils.{Constants, KafkaUtils}
import org.apache.flink.types.Nothing
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord, RecordMetadata}


object Server {

  private val gson: Gson = new GsonBuilder().create()

  def main(args: Array[String]): Unit = {
    //创建服务端socket
    val socket = new ServerSocket(8888)

    //创建客户端socket
    val clientSocket: Socket = socket.accept()
    val inputStream = new ObjectInputStream(clientSocket.getInputStream)
    val outputStream = new ObjectOutputStream(clientSocket.getOutputStream)

    //创建kafka连接
    val topicName: String = "test01"
    val kafkaProp: Properties = new Properties
    val sysProp: Properties = KafkaUtils.getSystemProp

    if (sysProp == null) {
      return
    }

    kafkaProp.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, sysProp.getProperty(Constants.KAFKA_PREFIX + ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))
    kafkaProp.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, sysProp.getProperty(Constants.KAFKA_PRODUCER_PREFIX + ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG))
    kafkaProp.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, sysProp.getProperty(Constants.KAFKA_PRODUCER_PREFIX + ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG))
    val producer: KafkaProducer[String, String] = new KafkaProducer[String, String](kafkaProp)


    //循环接受消息
    while (true) {

      val value: AnyRef = inputStream.readObject()
      value match {
        case LoggingData(id, logType, wellId, wellName, happenTime, _, logDepth, drillRate, wellDeflection, _) => {
          val data: LoggingData = LoggingData(id, logType, wellId, wellName, happenTime, System.currentTimeMillis(), logDepth, drillRate, wellDeflection, null)
          println(data)
          val record = new ProducerRecord[String, String]("socket_mudlog",wellName,
            gson.toJson(data))
          producer.send(record, (metadata: RecordMetadata, exception: Exception) => {
            println(record.timestamp())
          })
        }
        case _ => println("错误值！")
      }
    }

  }


}
